
#include <iostream>
#include <sstream>
#include <iomanip>

#include <boost/filesystem.hpp>
#include <boost/algorithm/string.hpp>

#include <boost/date_time/posix_time/posix_time.hpp>
using namespace boost::posix_time;

#include "bocat/common/ini.hpp"
#include "bocat/common/base64.hpp"
#include "bocat/common/time.hpp"
#include "bocat/common/logging.hpp"

#include "cubin.hpp"

namespace   sakcab   {


//static const char*  SNAPSHOT_ENCODE_base64 = "base64";


///////////////////////////////////////////////////////////////////////
//

Cubin::Cubin( const char* ini_filename, const char* ini_section )
		:   _config(ini_filename, ini_section),
			_oplog_p( NULL ),
			_index_default_p( NULL ),
			_id( ini_section ),
			_meter( _id )
{
	initialize( ini_filename, ini_section );

	_when_last_snapshot = 0;
}

Cubin::~Cubin()
{
	finalize();
}


void
Cubin::initialize( const char* ini_filename, const char* ini_section )
{
	bocat::common::INIFile	ini( ini_filename );

	// get count of idx
	std::string	idx_count = ini[ ini_section ][ "index.count" ];
	if (  idx_count.size() < 1 ) {
		std::ostringstream	msg;
		msg << "No option 'index.count' for [" << ini_section << "] in " << ini_filename;
		throw std::runtime_error( msg.str() );
	}

	std::string	default_dbhome = ini[ ini_section ][ "index.home" ];
	if ( default_dbhome.size() < 1 ) {
		std::ostringstream	msg;
		msg << "No option 'index.home' for [" << ini_section << "] in " << ini_filename;
		throw std::runtime_error( msg.str() );
	}

	boost::filesystem::path	default_dbhome_path( default_dbhome );
	if ( !boost::filesystem::exists( default_dbhome_path ) ) {
		try {
			boost::filesystem::create_directories( default_dbhome_path );
		}
		catch ( std::exception& e ) {
			std::ostringstream	msg;
			msg << "fail to create index.home " << default_dbhome << " -- " << e.what();
			throw	std::runtime_error( msg.str() );
		}
	}

	// get default setting for idx
	std::string	idx_default_section = ini[ ini_section ][ "index.setting" ];
	// override default dbhome
	ini[ idx_default_section ][ "index.home" ] = default_dbhome;

	std::ostringstream	stream_ini;
	stream_ini << ini[ idx_default_section ];
	_idx_default_settings = stream_ini.str();

	std::string	default_section = ini[ idx_default_section ][ "index.section" ];
	if ( default_section.size() < 1 ) {
		std::ostringstream	msg;
		msg << "No option 'index.section' for [" << idx_default_section << "] in " << ini_filename;
		throw std::runtime_error( msg.str() );
	}

	// create default bin
	std::string	default_db_filename = ini[ ini_section ][ "index.dbfile" ];
	if ( default_db_filename.size() < 1 ) {
		std::ostringstream	msg;
		msg << "No option 'index.dbfile' for [" << ini_section << "] in " << ini_filename;
		throw std::runtime_error( msg.str() );
	}
	std::string	default_ini_filename = default_db_filename + ".ini";
	boost::filesystem::path	default_ini_filepath( default_ini_filename );
	if ( !boost::filesystem::exists( default_ini_filepath ) ) {
		std::ostringstream	stream;
		stream << ini[ ini_section ];

		std::ofstream	ofs( default_ini_filename.c_str() );
		if ( !ofs ) {
			std::ostringstream	msg;
			msg << "fail to create " << default_ini_filename;
			throw std::runtime_error( msg.str() );
		}

		ofs << "[" << default_section << "]" << std::endl
			<< stream.str() << std::endl
			<< std::endl;

		ofs.close();
	}
	_index_default_p = new CuIndex( default_ini_filename.c_str(), default_section.c_str() );

	// create other bin
	boost::trim( idx_count );
	unsigned	idx_count_i = boost::lexical_cast< unsigned >( idx_count.c_str() );
	for (unsigned i=0; i<idx_count_i; ++i) {
		std::ostringstream	stream_db;
		stream_db << default_dbhome << "/" << std::setw(idx_count.size()) << std::setfill('0') << i << ".db";

		std::string	idx_db = stream_db.str();
		std::string	idx_ini = idx_db + ".ini";

		boost::filesystem::path	ini_path( idx_ini );
		if ( !boost::filesystem::exists( ini_path ) ) {
			std::ofstream	ofs( idx_ini.c_str() );
			if ( !ofs ) {
				std::ostringstream	msg;
				msg << "fail to create " << idx_ini;
				throw std::runtime_error( msg.str() );
			}

			ofs << "[" << default_section << "]" << std::endl
				<< "index.dbfile = " << idx_db << std::endl
				<< _idx_default_settings << std::endl
				<< std::endl;

			ofs.close();
		}

		index_ptr	idx_p = new CuIndex( idx_ini.c_str(), default_section.c_str() );
		_index_bins.push_back( idx_p );
	}

	// create snapshot home, if necessary
	if ( _config.index_snap_home.size() > 0 ) {
		boost::filesystem::path	snap_home_path( _config.index_snap_home );
		if ( !boost::filesystem::exists( snap_home_path ) ) {
			try {
				boost::filesystem::create_directories( snap_home_path );
			}
			catch ( std::exception& e ) {
				std::ostringstream	msg;
				msg << "fail to create index.snap.home " << _config.index_snap_home << " -- " << e.what();
				throw	std::runtime_error( msg.str() );
			}
		}
	}

	if ( _config.oplog_home.size() > 1 ) {
		boost::filesystem::path	oplog_home_path( _config.oplog_home );
		if ( !boost::filesystem::exists( oplog_home_path ) ) {
			try {
				boost::filesystem::create_directories( oplog_home_path );
			}
			catch ( std::exception& e ) {
				std::ostringstream	msg;
				msg << "fail to create oplog.home " << _config.oplog_home << " -- " << e.what();
				throw	std::runtime_error( msg.str() );
			}
		}

		_oplog_p = new OpLog(_config.oplog_home, _config.oplog_file_rotate_size, _config.oplog_file_slim, _config.oplog_key_encode.c_str(), _config.oplog_value_encode.c_str());
		if ( !_oplog_p->isClear() ) {
			// try to restore oplog
			bocat::common::Logger*  logger_p = bocat::common::getLogger();
			logger_p->info(__LINE__, __FILE__, "Cubin : oplog.home %s is not clear, try to recovery ...", _config.oplog_home.c_str() );

			_oplog_p->beginRecovery();

			while ( _oplog_p->hasMoreRecoveryIterator() ) {
				OpLogIterator*	it_p = _oplog_p->getNextRecoveryIterator();

				if ( NULL == it_p ) {
					continue;
				}

				Timestamp	ts;
				std::string op, kpin, key, data;
				while ( it_p->next(ts, op, kpin, key, data) > 0 ) {	
					index_ptr	index_p = getIndex( kpin, key );

					if ( OpLog::OP_PUT == op ) {
						index_p->put( key, data );
					}
					else if (OpLog::OP_DEL == op) {
						index_p->erase( key );
					}
				}
			}

			_oplog_p->endRecovery();

			logger_p->info(__LINE__, __FILE__, "Cubin : make indexes checkpoint ...");
			this->snapshot();
			logger_p->info(__LINE__, __FILE__, "Cubin : recovery oplog.home %s done.", _config.oplog_home.c_str() );

			//std::ostringstream	msg;
			//msg << "!!!TODO oplog.home " << _config.oplog_home << " is not clear, try to recovery ... ";
			//throw	std::runtime_error( msg.str() );
		}
	}
}


void
Cubin::finalize( void )
{
	bocat::common::Logger*  logger_p = bocat::common::getLogger();

	// take snapshot (check point)
	logger_p->info(__LINE__, __FILE__, "Cubin : make indexes checkpoint ...");
	this->snapshot();

	// remove index
	logger_p->info(__LINE__, __FILE__, "Cubin : close indexes ...");
	for (unsigned i=0; i < _index_bins.size(); ++i ) {
		index_ptr	p = _index_bins[ i ];
		delete p;
	}
	_index_bins.clear();

	if (NULL != _index_default_p) {
		delete _index_default_p;

		_index_default_p = NULL;
	}

	if (NULL != _oplog_p) {
		delete _oplog_p;

		_oplog_p = NULL;
	}

	logger_p->info(__LINE__, __FILE__, "Cubin : done.");
}


///////////////////////////////////////////////////////////////////////
//


int 
Cubin::insert( const std::string& kpin,
         const std::string& key,
         const std::string& data )
{
	boost::mutex::scoped_lock   _scoped_lock( _mutex );

	ptime   time_start = microsec_clock::local_time();

	Timestamp	ts;
	
	// log operation
	if ( NULL != _oplog_p ) {
		_oplog_p->log( ts, OpLog::OP_PUT, kpin, key, data );
	}

	std::string	value;
	value.reserve( sizeof(Timestamp) + data.size() );
	value.append( (const char*)&ts, sizeof(Timestamp) );
	value.append( data );

	index_ptr	index_p = getIndex( kpin, key );
    index_p->put( key, value );

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countInsert( time_last );

    return  0;
}


int 
Cubin::insertAppend( const std::string& kpin,
         const std::string& key,
         const std::string& data )
{
	boost::mutex::scoped_lock   _scoped_lock( _mutex );

	ptime   time_start = microsec_clock::local_time();

	Timestamp	ts;
	
	// log operation
	if ( NULL != _oplog_p ) {
		_oplog_p->log( ts, OpLog::OP_PUT, kpin, key, data );
	}

	std::string	value;
	value.reserve( sizeof(Timestamp) + data.size() );
	value.append( (const char*)&ts, sizeof(Timestamp) );
	value.append( data );

	index_ptr	index_p = getIndex( kpin, key );
    index_p->put( key, value );

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countInsert( time_last );

    return  0;
}


int 
Cubin::insertIgnore( const std::string& kpin,
         const std::string& key,
         const std::string& data )
{
	boost::mutex::scoped_lock   _scoped_lock( _mutex );

	ptime   time_start = microsec_clock::local_time();

	index_ptr	index_p = getIndex( kpin, key );

    // STEP. check if key exists
    //std::string skey = getSkeyByKpin( kpin );
    std::string	data_old;
    try {
        //index_p->get( skey, data_old );
        index_p->get( key, data_old );
    }
    catch( std::exception& e ) {
        std::ostringstream  msg;
        msg << "fail to check existence for key '" << key << "' -- " << e.what();
        throw std::runtime_error( msg.str() );
    }

    // STEP. if key exists, ignore it
    if ( data_old.size() > 0 ) {
        return 0;
    }

	Timestamp	ts;
	
	// log operation
	if ( NULL != _oplog_p ) {
		_oplog_p->log( ts, OpLog::OP_PUT, kpin, key, data );
	}

	std::string	value;
	value.reserve( sizeof(Timestamp) + data.size() );
	value.append( (const char*)&ts, sizeof(Timestamp) );
	value.append( data );

    index_p->put( key, value );

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countInsert( time_last );

    return  0;
}


int 
Cubin::find( const std::string& kpin,
         const std::string& key,
               std::string& data )
{   
	boost::mutex::scoped_lock   _scoped_lock( _mutex_read );

	ptime   time_start = microsec_clock::local_time();

	index_ptr	index_p = getIndex( kpin, key );

    // STEP. check if key exists
    try {
		std::string	value;
        index_p->get( key, value );

		if ( value.size() < sizeof(Timestamp) ) {
			return	0;
		}
		data.assign( value.c_str() + sizeof(Timestamp), value.size() - sizeof(Timestamp) );
    }
    catch( std::exception& e ) {
        std::ostringstream  msg;
        msg << "fail to check existence for key '" << key << "' -- " << e.what();
        throw std::runtime_error( msg.str() );
    }

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countFind( time_last );

    // STEP. if key does NOT exist, just return
    if ( data.size() < 1 ) {
        return 0;
    }

    return  data.size();
}


int
Cubin::find( const std::string& kpin,
             const std::string& key,
                   std::vector< std::string >& all_data,
                   u_int32_t limit )
{
    boost::mutex::scoped_lock   _scoped_lock( _mutex_read );

	ptime   time_start = microsec_clock::local_time();

	index_ptr	index_p = getIndex( kpin, key );

    // STEP. check if key exists
	std::string	data;
    try {
		std::string	value;
        index_p->get( key, value );

		if ( value.size() < sizeof(Timestamp) ) {
			return	0;
		}
		data.assign( value.c_str() + sizeof(Timestamp), value.size() - sizeof(Timestamp) );
    }
    catch( std::exception& e ) {
        std::ostringstream  msg;
        msg << "fail to check existence for key '" << key << "' -- " << e.what();
        throw std::runtime_error( msg.str() );
    }

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countFind( time_last );

    // STEP. if key does NOT exist, just return
    if ( data.size()  < 1 ) {
        return 0;
    }

	all_data.push_back( data );

    return  all_data.size();
}



///////////////////////////////////////////////////////////////////////
//

int 
Cubin::erase( const std::string& kpin,
           const std::string& key )
{
	boost::mutex::scoped_lock   _scoped_lock( _mutex );

	ptime   time_start = microsec_clock::local_time();

	index_ptr	index_p = getIndex( kpin, key );

    //std::string skey = getSkeyByKpin( kpin );
    //index_p->erase( skey );
    index_p->erase( key );

	// log operation
	if ( NULL != _oplog_p ) {
		Timestamp	ts;

		_oplog_p->log( ts, OpLog::OP_DEL, kpin, key, "X" );
	}

	ptime   time_stop = microsec_clock::local_time();
	long	time_last = getTimeDuration( time_start, time_stop );
	_meter.countErase( time_last );

    return  0;
}


///////////////////////////////////////////////////////////////////////
//

int
Cubin::insertAppend( const std::vector< std::string >& kpin_v,
					 const std::vector< std::string >& key_v,    
					 const std::vector< std::string >& data_v )
{
	boost::mutex::scoped_lock   _scoped_lock( _mutex );


	Timestamp	ts;
	
	for (size_t k=0; k<key_v.size(); ++k) {
		const std::string&	kpin = kpin_v[ k ];
		const std::string&	key  = key_v[ k ];
		const std::string&	data = data_v[ k ];

		// log operation
		if ( NULL != _oplog_p ) {
			_oplog_p->log( ts, OpLog::OP_PUT, kpin, key, data );
		}

		std::string	value;
		value.reserve( sizeof(Timestamp) + data.size() );
		value.append( (const char*)&ts, sizeof(Timestamp) );
		value.append( data );

		index_ptr	index_p = getIndex( kpin, key );
		index_p->put( key, value );

	}

    return  0;
}


///////////////////////////////////////////////////////////////////////
//


int
Cubin::slim( void )
{
    boost::mutex::scoped_lock   _scoped_lock( _mutex );

	bocat::common::Logger*  logger_p = bocat::common::getLogger();

	// slim operation logs
	if ( _config.data_home.size() > 1 ) {
		OpLog	oplog(_config.data_home, _config.oplog_file_rotate_size, _config.oplog_file_slim, _config.oplog_key_encode.c_str(), _config.oplog_value_encode.c_str());

		OpLogIterator*  it_p = NULL;
		int     count_slimmed = 0;

		while( (it_p = oplog.getSlimOpLogIterator()) != NULL ) { 
			try {
				logger_p->info(__LINE__, __FILE__, " slimming %s ...", oplog.getSlimmingFile().c_str());

				Timestamp	ts;
				std::string op, kpin, key, data_logged;
				//int		fileno_slimming = oplog.getSlimmingFileNo();
				int		count_slim_each_file = 0;
				while ( it_p->next(ts, op, kpin, key, data_logged) > 0 ) {	
					index_ptr	index_p = getIndex( kpin, key );

					// STEP. check if key exists
					std::string	data;
					try {
						index_p->get( key, data );
					}
					catch( std::exception& e ) {
						// just ignore ...
						//std::ostringstream  msg;
						//msg << "fail to check existence for key '" << key << "' -- " << e.what();
						////throw std::runtime_error( msg.str() );
						//logger_p->error( __LINE__,__FILE__, "SLIM ERROR : %s", msg.str().c_str() );
						continue;
					}

					timeval_type	tv;
					if ( data.size() < sizeof(timeval_type) ) {
						continue;
					}
					memcpy( &tv, data.c_str(), sizeof(timeval_type) );

					// check if fileno matched
					if ( OpLog::OP_PUT == op ) {
						if ( ts != tv ) {
							continue;
						}
					}
					else if (OpLog::OP_DEL == op) {
						if ( ts < tv ) {
							// erase operation occurs before current key insertion
							continue;
						}
					}

					index_p->erase( key );

					++count_slim_each_file;
					++count_slimmed; 
				}
				logger_p->info(__LINE__,__FILE__," slim %s done (%d records slimmed).", oplog.getSlimmingFile().c_str(), count_slim_each_file);
			}
			catch( std::exception& e ) {
				logger_p->info(__LINE__,__FILE__," slim %s failed.", oplog.getSlimmingFile().c_str());
				logger_p->error(__LINE__,__FILE__," slim exception : %s", e.what());
				logger_p->error(__LINE__,__FILE__," slim %s failed.", oplog.getSlimmingFile().c_str());

				oplog.freeSlimOpLogIterator( it_p, false );
				continue;
			}       

			oplog.freeSlimOpLogIterator( it_p );
		}       
	}

	// optimize index
	logger_p->info(__LINE__, __FILE__, " optimizing index %s ...", _id.c_str());
	for (unsigned i=0; i < _index_bins.size(); ++i ) {
		logger_p->info(__LINE__, __FILE__, "  - optimizing #%u index ...", i);
		index_ptr	p = _index_bins[ i ];

		p->optimize();
	}

	if (NULL != _index_default_p) {
		logger_p->info(__LINE__, __FILE__, "  - optimizing default index ...");
		_index_default_p->optimize();
	}
	logger_p->info(__LINE__, __FILE__, " optimize index %s done.", _id.c_str());

	// snap-shot
	this->snapshot();

	return	0;
}


int
Cubin::snapshot( void )
{
	bocat::common::Logger*  logger_p = bocat::common::getLogger();

	time_t	now = ::time( NULL );
	if ( (_config.index_snap_home.size() > 0) && 
		 (
		 	((now - _when_last_snapshot) > _config.index_snap_interval)  ||
			(
				(NULL != _oplog_p) && (_oplog_p->getTankage() >= _config.oplog_file_tankage)
			)
		 )
	   ) {
		logger_p->info(__LINE__, __FILE__, " snapshoting index %s ...", _id.c_str());
		for (unsigned i=0; i < _index_bins.size(); ++i ) {
			char	filename[ 512 ], fn_prefix[ 32 ];
			if ( _index_bins.size() < 10 ) {
				snprintf( filename, sizeof(filename) - 1, "%s/%u.snap-%lu", _config.index_snap_home.c_str(), i, now );
				snprintf( fn_prefix, sizeof(fn_prefix) - 1, "%u.snap-", i );
			}
			else {
				snprintf( filename, sizeof(filename) - 1, "%s/%02u.snap-%lu", _config.index_snap_home.c_str(), i, now );
				snprintf( fn_prefix, sizeof(fn_prefix) - 1, "%02u.snap-", i );
			}
			logger_p->info(__LINE__, __FILE__, "  - snapshoting --> %s ...", filename);
			index_ptr	p = _index_bins[ i ];

			int	count_records = p->snapshot( filename, _config.index_snap_key_encode.c_str(), _config.index_snap_value_encode.c_str() );

			checkSnapshotCopies( _config.index_snap_home.c_str(), fn_prefix, _config.index_snap_copies );

			logger_p->info(__LINE__, __FILE__, "    snapshot %s done (%d records).", filename, count_records);
		}

		if (NULL != _index_default_p) {
			char	filename[ 512 ], fn_prefix[ 32 ];
			snprintf( filename, sizeof(filename) - 1, "%s/default.snap-%lu", _config.index_snap_home.c_str(), now );
			snprintf( fn_prefix, sizeof(fn_prefix) - 1, "default.snap-" );
			logger_p->info(__LINE__, __FILE__, "  - snapshoting --> %s ...", filename);

			int	count_records = _index_default_p->snapshot( filename, _config.index_snap_key_encode.c_str(), _config.index_snap_value_encode.c_str() );

			checkSnapshotCopies( _config.index_snap_home.c_str(), fn_prefix, _config.index_snap_copies );

			logger_p->info(__LINE__, __FILE__, "    snapshot %s done (%d records).", filename, count_records);
		}
		logger_p->info(__LINE__, __FILE__, " snapshot index %s done.", _id.c_str());

		// after snap-shot, all oplog-files may be cleared
		if ( NULL != _oplog_p ) {
			logger_p->info(__LINE__, __FILE__, " - oplog files tankage %d Bytes.", _oplog_p->getTankage());
			if ( _config.oplog_file_slim < 1 ) {
				logger_p->info(__LINE__, __FILE__, " - clear all oplog files till this checkpoint ..." );
				int	count_file_cleared;
				count_file_cleared = _oplog_p->clearDataDirectory();
				logger_p->info(__LINE__, __FILE__, "   clear all oplog files till this checkpoint done (%d files droped).", count_file_cleared );
			}
			else {
				logger_p->info(__LINE__, __FILE__, " - keep all oplog files till this checkpoint to slim ..." );
				int	count_file_cleared;
				if ( _config.data_home.size() > 1 ) {
					count_file_cleared = _oplog_p->clearDataDirectory( _config.data_home.c_str() );
					logger_p->info(__LINE__, __FILE__, "   move all oplog files till this checkpoint done (%d files to slim).", count_file_cleared );
				}
				else {
					count_file_cleared = _oplog_p->clearDataDirectory();
					logger_p->info(__LINE__, __FILE__, "   No data-home defined, clear all oplog files till this checkpoint done (%d files droped).", count_file_cleared );
				}
			}
		}

		_when_last_snapshot = ::time(NULL);
	}

	return	0;
}


///////////////////////////////////////////////////////////////////////
//


const std::string   
Cubin::getSkeyByKpin( const std::string& kpin )
{
	char	skey[32];
	u_int32_t	skey_size = _config.index_skey_size;

	memset(skey, 0, sizeof(skey));
	if ( kpin.size() < skey_size ) {
		skey_size = kpin.size();
	}

	memcpy( skey, kpin.c_str() + (kpin.size() - skey_size), skey_size );

	return	std::string( skey, skey_size );
}


Cubin::index_ptr
Cubin::getIndex( const std::string& kpin, const std::string key )
{
	if ( _index_bins.size() < 1 ) {
		return	_index_default_p;
	}

	// assert( kpin.size() == 16 );

	const char*	p = kpin.c_str();
	size_t	s = kpin.size();
	if ( s < 16 ) {
		std::ostringstream	msg;
		msg << "BAD kpin (size=" << s << ") for KEY: " << key;

		throw std::runtime_error( msg.str() );
	}

	int	x =  (p[7] << 8) + p[8];
	return	_index_bins[ x % _index_bins.size() ];
}

///////////////////////////////////////////////////////////////////////
//


}   //// namespace sakcab

